package org.dormi.learn.completablefuture;

import java.util.concurrent.*;

/**
 * CompletableFuture | thenApply vs thenCompose
 * https://stackoverflow.com/questions/43019126/completablefuture-thenapply-vs-thencompose
 * <p>
 * Do you understand the difference between map and flatMap in Stream?
 * thenApply is the map and thenCompose is the flatMap of CompletableFuture.
 * You use thenCompose to avoid having CompletableFuture<CompletableFuture<..>>
 * <p>
 * thenCompose() 使用上一个 CompletableFuture 的返回, 并构建一个新的 CompletableFuture, 并做为返回值
 * CompletableFuture f1 -> CompletableFuture f2 -> ...
 * f1 可以被不同的 Executor执行
 * <p>
 * thenApply() 只对上一个 CompletableFuture 的值进行处理, 还在同一个 Executor中
 * CompletableFuture<CompletableFuture<T>>>
 * 像嵌套
 *
 * Returns a new CompletionStage that, when this stage completes normally, is executed with this stage as the argument to the supplied function.
 * Returns a new CompletionStage that, when this stage completes normally, is executed with this stage's result as the argument to the supplied function.
 *
 * @author : dormi330
 * @since : 22/04/18
 */
public class CompletableFutureDemo04 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("x1 " + Thread.currentThread().getName());
            timeConsuming(3000);
            return 30;
        }, executorService)
            .thenCompose(money -> CompletableFuture.supplyAsync(() -> {
                System.out.println("x2 " + Thread.currentThread().getName());
                Integer point = money < 50 ? 1 : 2;
                return point;
            }, executorService))
            .thenCompose(point -> CompletableFuture.supplyAsync(() -> {
                System.out.println("x3 " + Thread.currentThread().getName());
                Integer pointPlus = point + 1;
                return pointPlus;
            }, executorService))
            .thenCompose(point -> CompletableFuture.supplyAsync(() -> {
                System.out.println("x4 " + Thread.currentThread().getName());
                Integer pointPlus = point + 1;
                return pointPlus;
            }, executorService))
            .thenCompose(point -> CompletableFuture.supplyAsync(() -> {
                System.out.println("x5 " + Thread.currentThread().getName());
                Integer pointPlus = point + 1;
                return pointPlus;
            }, executorService));

        try {
            Integer result = future.get(5, TimeUnit.SECONDS);
            System.out.println("result= " + result);//result=1
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }

        executorService.shutdown();
    }
    /*
x1 pool-1-thread-1
x2 pool-1-thread-2
x3 pool-1-thread-3
x4 pool-1-thread-4
x5 pool-1-thread-5
result= 4
    */

    /** 模拟耗时处理 */
    private static void timeConsuming(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
